/*
 * ===================================================================
 *  
 *      Filename: log_srv_policy.c
 *
 *   Description: This is the main entry for the policy modules.The policy
 *                module is a daemon thread,which listens on the local
 *                socket,if we want to update the policy file,we need to
 *                modify the policy file firstly in the server, then exe
 *                the update script.
 *                The server will only update the policy to the clients 
 *                which sent the logs to them.(known client)
 *
 *       Version:  1.0
 *        Create:  2010/07/13
 *
 *        Author:  Keven Zhu     kevenzhu@tencent.com
 *     Copyright:  Reserved by Tencent.com Co.ltd
 *   
 * ====================================================================
 */
#include <stdio.h>
#include <stdlib.h>
#define  __USE_GNU
#include <string.h>
#include <errno.h>
#include <netdb.h>
#include <time.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/un.h>
#include <signal.h>
#include <sched.h>
#include <pthread.h>
#include <sys/param.h>
#include <linux/unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include "lib/util.h"

/*globel policy pthread id*/
pthread_t pid;
pthread_t flush_pid;

/*max policy message size*/
#define MAX_POLICY_SIZE     100
#define CLIENT_DAEMON_PORT  41501

/*does the update have finished, if yes,allowing the next update, otherwise pending*/
int update_policy_in_processing = 0;

struct remote_client_hash_list
{
    struct hlist_head    head;
    pthread_spinlock_t   lock;
};

struct remote_client_hash_info
{
    struct hlist_node   list;
    uint                addr;
    uint                time;
};

/* we choose hlist for the client info, and we assume that
 * most of the client is located in the same network
 */
#define HASH_MASK_CLIENT  255
struct  remote_client_hash_list policy_remote_client_list[256];

/*we only support the ipv4 clinet*/
int add_remote_client(uint addr)
{
    struct hlist_node *loop_node;
    struct remote_client_hash_info *tmp;
    struct remote_client_hash_info *loop_container;
    struct remote_client_hash_list *tmp_head = &policy_remote_client_list[addr>>24/*&HASH_MASK_CLIENT*/];
 
    pthread_spin_lock(&tmp_head->lock);
    /*it's the first node*/
    if(hlist_empty(&tmp_head->head))
    {
        tmp = (struct remote_client_hash_info *)malloc(sizeof(struct remote_client_hash_info));
        if(tmp)
        {
            tmp->addr = addr;
            tmp->time = time(NULL);
            hlist_add_head(&tmp->list,&tmp_head->head);
        }
        else
        {
            printf("No memory for the hlist node, exit\n");
        }
    }
    /*to test confict or exist one*/
    else
    {
        hlist_for_each_entry(loop_container,loop_node,&tmp_head->head,list)
        {
            if(loop_container->addr == addr)
            {
                pthread_spin_unlock(&tmp_head->lock);
                return OK;
            }
        }
        tmp = (struct remote_client_hash_info *)malloc(sizeof(struct remote_client_hash_info));
        tmp->addr = addr;
        tmp->time = time(NULL);
        hlist_add_head(&tmp->list,&tmp_head->head);
    }
    pthread_spin_unlock(&tmp_head->lock);
    return OK;
}

/*add an client into the list*/
int remove_remote_client(uint addr)
{
    struct hlist_node *loop_node,  *n_node;
    struct remote_client_hash_info *loop_container;
    struct remote_client_hash_list *tmp_head = &policy_remote_client_list[addr>>24/*&HASH_MASK_CLIENT*/];

    pthread_spin_lock(&tmp_head->lock);
    hlist_for_each_entry_safe(loop_container,loop_node,n_node,&tmp_head->head,list)
    {
        if(loop_container->addr == addr)
        {
            hlist_del(&loop_container->list);
            pthread_spin_unlock(&tmp_head->lock);
            free((char*)loop_container);
            return OK;
        }
    }
    pthread_spin_unlock(&tmp_head->lock);
    return FALSE;
}

unsigned int flush_client_list()
{
    #define MAX_TO_VAL  1800 /* 60*30 seconds*/
    /*the client list is obsoluted*/
    struct hlist_node *loop_node,  *n_node;
    struct remote_client_hash_info *loop_container;
    struct remote_client_hash_list *tmp_head = (struct remote_client_hash_list *)&policy_remote_client_list;
    int i = 0;
    int sec = time(NULL);
    
    /*free the client nodes*/
    for(i =0;i<256;i++)
    {
        /*the thread has been terminated,we can delete the list directly*/
        pthread_spin_lock(&tmp_head->lock);
        hlist_for_each_entry_safe(loop_container,loop_node,n_node,&tmp_head->head,list)
        {
            if((sec - loop_container->time) > MAX_TO_VAL)
            {
                hlist_del(&loop_container->list);
                free((char*)loop_container);
            }
        }
        pthread_spin_unlock(&tmp_head->lock);
        tmp_head++;
    }
    return OK;
}

/*be responsibility for recving/sending policy update msg*/
void *flush_routine(void *argv)
{
    while(1)
    {
        if(update_policy_in_processing)
        {
            /*sleep 5 minutes*/
            sleep(5*60);
        }
        else
        {
            flush_client_list();
            sleep(60*30);
        }
    }
}

/*send the policy fiel to the listening client one by one*/
int update_policy_file(char *file_path)
{
    struct stat file_status;
    char   *buff,*tmp_buff,*msg_buff,*tmp_msg_buff;
    char   databuff[MAX_LOG_SIZE];
    int    icounter,fd,sockfd,iret;
    struct timeval tv;
    struct sockaddr_in client_addr,tmp_addr;    
    struct hlist_node *loop_node,  *n_node;
    struct remote_client_hash_info *loop_container;
    struct remote_client_hash_list *tmp_head = (struct remote_client_hash_list *)&policy_remote_client_list;
 

    if(stat(file_path,&file_status))
    {
        perror("failed to get the file inf");
        update_policy_in_processing = FALSE; 
        return FALSE;
    }

    buff = (char *)malloc(file_status.st_size);
    if(!buff)
    {
        printf("failed to get the memory for polict\n");
        update_policy_in_processing = FALSE; 
        return FALSE;
    }

    /*during the update stage, new connected client will NOT be notified!!!!!!*/
    if(-1 == (fd = open(file_path,O_RDONLY)))
    {
        free(buff);
        update_policy_in_processing = FALSE; 
        perror("failed to open the file");
        return FALSE;
    }
    tmp_buff = buff;
    iret = 0;
    while((icounter = read(fd,tmp_buff,SSIZE_MAX))>0)
    {
        tmp_buff = tmp_buff + icounter;
        iret = iret + icounter;
    }
    icounter = iret;
    
    /*malloc buffer for the msg packet*/
    msg_buff = (char*)malloc(64);
    if(!msg_buff)
    {
        update_policy_in_processing = FALSE;
        free(buff);
        printf("failed to get the memory\n");
        return FALSE;
    }
    msg_buff[63]= '\0';

    /*construct the policy update request*/
    msg_buff[0] = (char)((CS_PROTOCOL_POLICY<<4)|CS_VERSION);
    msg_buff[1] = UPDATE_POLICY_REQUEST; 

    /*now we get all of the msg in the buff*/
    tmp_msg_buff = strcasestr(buff,"version");
    if(tmp_msg_buff)
    {
        tmp_msg_buff = tmp_msg_buff + 7;//strlen("version")
        tmp_msg_buff = strcasestr(tmp_msg_buff,"version");
        if(tmp_msg_buff)
        {
            while(*tmp_msg_buff != '=')
                tmp_msg_buff++;
            tmp_msg_buff++;
            *((int*)(msg_buff+2)) = atoi(tmp_msg_buff);
        }
        else
            msg_buff[2]=0x1;

    }
    else
        msg_buff[2]=0x1;

    /*loop the client and push the policy*/
    for(iret =0;iret<256;iret++)
    {
        /*the thread has been terminated,we can delete the list directly*/
        pthread_spin_lock(&tmp_head->lock);
        hlist_for_each_entry_safe(loop_container,loop_node,n_node,&tmp_head->head,list)
        {    
            /*bind to the local unix socket*/
            sockfd = socket(AF_INET,SOCK_DGRAM,0);
            if(-1 == sockfd)
            {
                perror("error to create the client socket for policy");
                free(buff);
            }

            /*set the recv timeout value*/
            tv.tv_sec = 2;
            tv.tv_usec= 0;
            setsockopt(sockfd,SOL_SOCKET,SO_RCVTIMEO,&tv,sizeof(tv));  

            /* build address */
            memset((void*)&client_addr, 0, sizeof(client_addr));
            client_addr.sin_family = AF_INET;
            client_addr.sin_addr.s_addr = loop_container->addr;//inet_addr(POLICY_MULT_GROUP); /* multicast group ip */ 
            client_addr.sin_port = htons(CLIENT_PORT);
            iret = sendto(sockfd,msg_buff,10,0,(struct sockaddr*)(&client_addr),sizeof(client_addr));
            iret = recvfrom(sockfd,msg_buff,10,0,(struct sockaddr*)(&tmp_addr),(socklen_t*)&iret);
            if(iret >0)
            {
                if(msg_buff[1] == UPDATE_POLICY_ACCEPT)
                {
                    /*agree with the update,continue*/
                }
                else
                {
                    close(sockfd);
                    continue;
                }
            }
            else
            {
                perror("error to get the reply");
                /*recv error,ignore this client*/
                close(sockfd);
                continue;
            }
    
            /****************************************************************************************/
            databuff[0] = (CS_PROTOCOL_POLICY<<4)|CS_VERSION;
            databuff[1] = UPDATE_POLICY_DATA;
            /*we got the client's reply,now prepare to send the policy data*/
            tmp_buff = buff;
            while(icounter)
            {
                iret = (icounter<(MAX_LOG_SIZE-2))?icounter:(MAX_LOG_SIZE-2);
                memcpy(databuff+2,tmp_buff,iret);
                if(-1 == sendto(sockfd,databuff,iret+2,0,(struct sockaddr*)(&client_addr),sizeof(client_addr)))
                {
                    printf("Failed to update the policy\n");
                    close(sockfd);
                    continue;
                }
                tmp_buff = tmp_buff + iret;
                icounter = icounter - iret;
            }
            /*send to finish msg to the client*/
            databuff[1] = UPDATE_POLICY_FINISH;
            /*send twice to make sure the client has recv the command*/
            sendto(sockfd,databuff,2,0,(struct sockaddr*)(&client_addr),sizeof(client_addr));
            sendto(sockfd,databuff,2,0,(struct sockaddr*)(&client_addr),sizeof(client_addr));
            /****************************************************************************************/
        }
        pthread_spin_unlock(&tmp_head->lock);
        tmp_head++;
    }            
    free(buff);
    free(msg_buff);
    update_policy_in_processing = FALSE;
    return OK; 
}

/*be responsibility for recving/sending policy update msg*/
void *policy_routine(void *argv)
{
    int    iret,msg_size,fd,remote_size,client_fd;
    struct sockaddr_in serv_addr,remote_addr;
    char   buff[MAX_POLICY_SIZE],*tmp;

    /*bind to the local unix socket*/
    fd = socket(AF_INET,SOCK_STREAM,0);
    if(-1 == fd)
    {
        perror("failed to create the local socket");
        return FALSE;
    }

    serv_addr.sin_family = AF_INET;
    serv_addr.sin_port   = htons(MNG_PORT);
    serv_addr.sin_addr.s_addr   = htonl(INADDR_ANY);//*((struct in_addr*)host->h_addr);
    //bzero(&(serv_addr.sin_zero),8);
    
    bind(fd,(struct sockaddr*)&serv_addr,sizeof(serv_addr));

    /*protocol for the reopen file(color/log),update policy file*/
    /*|----------------------------------------------------------------
     *|action(byte)|lenth(byte)(optional)| msg body(file name)        |
     *|---------------------------------------------------------------|
     *|                    msg(file name)                             |
     *|---------------------------------------------------------------|
     *action:
     *0x01   update policy file
     *0x02   reopen log   file
     *0x03   reopen color file
     *0x04   accept the request
     *0x05   reject the request,if we reject the request, rejecting reason will be filled in the msg body
     */
    listen(fd,50);
    remote_size = sizeof(struct sockaddr_in);
    while(OK)
    {
        client_fd = accept(fd,(struct sockaddr*)&remote_addr,(socklen_t *)&remote_size);
        if(-1 == client_fd)
        {
            perror("policy module: error to accept the new socket");
            continue;
        }
        memset(buff,0,MAX_POLICY_SIZE);//make sure the default value is zero
        iret = recv(client_fd,buff,MAX_POLICY_SIZE,0);
        if(iret > 0)
        {
            tmp = buff;
            switch(*tmp)
            {
                case UPDATE_POLICY_FILE:
                    if(update_policy_in_processing) //suppose we support async, but NOT now
                    {
                        iret = FALSE;
                    }
                    else
                    {
                        /*default policy file, no given parameter*/
                        if(1 == iret)
                        {
                            iret = update_policy_file(DEFAULT_POLICY_PATH);
                        }
                        else
                        {
                            update_policy_in_processing = TRUE;
                            tmp++;                //shift to the msg length
                            msg_size = *tmp;
                            buff[msg_size+2]='\0';//add the '\0' to terminate the string
                            tmp++;                //shift to the msg body
                            iret = update_policy_file(tmp);//will block all the new msg util update is finished
                        }
                    }
                    break;
                case REOPEN_LOG_FILE:
                    if(1 == iret) /*1 means only the action is sent, no file path*/
                    {
                        iret = reopen_local_file(REOPEN_LOG_FILE,sizeof(DEFAULT_LOG_PATH),DEFAULT_LOG_PATH);
                    }
                    else
                    {
                        tmp++;
                        msg_size = *tmp;
                        buff[msg_size+2]='\0';
                        tmp++;
                        iret = reopen_local_file(REOPEN_LOG_FILE,msg_size,tmp);
                    }
                    break;
                case REOPEN_COLOR_FILE:
                    if(1 == iret) /*1 means only the action is sent,no file path*/
                    {
                        iret = reopen_local_file(REOPEN_COLOR_FILE,sizeof(DEFAULT_COLOR_PATH),DEFAULT_COLOR_PATH);
                    }
                    else
                    {
                        tmp++;
                        msg_size = *tmp;
                        buff[msg_size+2]='\0';
                        tmp++;
                        iret = reopen_local_file(REOPEN_COLOR_FILE,msg_size,tmp);
                    }
                    break;
                default:
                    printf("unexpected msg,ignore\n");
                    continue;
            }
            /*send the rely message to the client*/
            tmp = buff;
            if(FALSE == iret)
                *tmp = REJECT_POLICY_REQUEST;
            else
                *tmp = ACCEPT_POLICY_REQUEST;
            
            send(client_fd,buff,1,0); // :)
            
        }
        else
        {
            perror("error in receiving the policy local socket");
        }
    }
}

/*initialize the policy modules*/
int init_tlogd_policy()
{
    int i = 0;
    /*we didn't keep the client list,the mulcast is used*/
    struct    remote_client_hash_list *tmp_head = (struct remote_client_hash_list *)&policy_remote_client_list;
    
    for(i =0;i<256;i++)
    {
        INIT_HLIST_HEAD(&tmp_head->head);
        if(pthread_spin_init(&tmp_head->lock,PTHREAD_PROCESS_PRIVATE))
        {
            perror("failed to initialize the spin lock");
            return FALSE;
        }
        tmp_head++;
    }

    if(pthread_create(&flush_pid,NULL,flush_routine,NULL))
    {
        perror("failed to create the policy thread");
        return FALSE;
    }
    if(pthread_create(&pid,NULL,policy_routine,NULL))
    {
        perror("failed to create the policy thread");
        return FALSE;
    }
    pthread_detach(pid);
    pthread_detach(flush_pid);

    return OK;
}

/*destroy the policy modules*/
int finit_tlogd_policy()
{
    /*the client list is obsoluted*/
    struct hlist_node *loop_node,  *n_node;
    struct remote_client_hash_info *loop_container;
    struct remote_client_hash_list *tmp_head = (struct remote_client_hash_list *)&policy_remote_client_list;
    int i = 0;
    
    printf("    stop the policy thread\n");
    /*destroy the policy pthread*/
    pthread_cancel(pid);
    
    printf("    stop the flush thread\n");
    /*destroy the policy pthread*/
    pthread_cancel(flush_pid);

    printf("    free the policy hash list\n");
    /*free the client nodes*/
    for(i =0;i<256;i++)
    {
        /*the thread has been terminated,we can delete the list directly*/
        //pthread_spin_lock(&tmp_head->lock);
        hlist_for_each_entry_safe(loop_container,loop_node,n_node,&tmp_head->head,list)
        {
            hlist_del(&loop_container->list);
            free((char*)loop_container);
        }
        //pthread_spin_unlock(&tmp_head->lock);
        pthread_spin_destroy(&tmp_head->lock);
        tmp_head++;
    }
    printf("    success to stop the policy modules\n");
    return OK;
}

